package com.mark.article.kafka.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.mark.article.service.AppArticleService;
import com.mark.common.kafka.KafkaListener;
import com.mark.common.kafka.KafkaTopicConfig;
import com.mark.common.kafka.messages.app.ArticleVisitStreamMessage;
import com.mark.model.mess.app.ArticleVisitStreamDto;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;


/**
 * @Description :  文章增量处理侦听器
 * @Author :  Markburt
 * @CreateDate :  2020/2/28$ 下午 07:23$
 * @UpdateUser :  Markburt
 * @UpdateDate :  2020/2/28$ 下午 07:23$
 * @UpdateRemark :  Project Build
 * @Version :  1.0
 */

@Component
@Log4j2
public class ArticleIncrHandleListener implements KafkaListener<String,String> {

    @Autowired
    private KafkaTopicConfig kafkaTopicConfig;
    @Autowired
    private ObjectMapper mapper;

    @Autowired
    private AppArticleService appArticleService;

    @Override
    public String topic() {
        return kafkaTopicConfig.getArticleIncrHandle();
    }

    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Consumer<?, ?> consumer) {
        log.info("接收到的消息为:{}", consumerRecord);
        String message =consumerRecord.value();
        try {
            ArticleVisitStreamMessage messages = mapper.readValue(message, ArticleVisitStreamMessage.class);
            ArticleVisitStreamDto dto = messages.getData();
            appArticleService.updateArticleView(dto);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
